package net.xdclass.listener;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import net.xdclass.config.KafkaTopicConfig;
import net.xdclass.req.ReportUpdateReq;
import net.xdclass.service.ReportDetailService;
import net.xdclass.service.ReportService;
import net.xdclass.util.JsonUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 * @author 小二
 * 监听Kafka的监听器
 */
@Slf4j
@Component
public class MQListener {

    @Resource
    private ReportDetailService reportDetailService;

    @Resource
    private ReportService reportService;

    /**
     * 消费监听，压测日志详情
     * @param record
     * @param ack
     * @param topic
     */
    @KafkaListener(topics = {KafkaTopicConfig.STRESS_TOPIC_NAME}, groupId = "stress-test-gp")
    public void onStressReportDetailMessage(ConsumerRecord<?,?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.warn("消息主题： {}， 分区： {}， 消息： {}", record.topic(), record.partition(), record.value().toString());

        reportDetailService.handleStressReportDetail(record.value().toString());
        // 消费
        ack.acknowledge();
    }

    /**
     * 消费监听，接口测试日志详情
     * @param record
     * @param ack
     * @param topic
     */
    @KafkaListener(topics = {KafkaTopicConfig.API_TOPIC_NAME}, groupId = "api-test-gp")
    public void onApiReportDetailMessage(ConsumerRecord<?,?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.warn("消息主题： {}， 分区： {}， 消息： {}", record.topic(), record.partition(), record.value().toString());

        reportDetailService.handleApiReportDetail(record.value().toString());
        // 消费
        ack.acknowledge();
    }

    /**
     * 消费监听，UI测试日志详情
     * @param record
     * @param ack
     * @param topic
     */
    @KafkaListener(topics = {KafkaTopicConfig.UI_TOPIC_NAME}, groupId = "ui-test-gp")
    public void onUiReportDetailMessage(ConsumerRecord<?,?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        log.warn("消息主题： {}， 分区： {}， 消息： {}", record.topic(), record.partition(), record.value().toString());

        reportDetailService.handleUiReportDetail(record.value().toString());
        // 消费
        ack.acknowledge();
    }



    /**
     * 消费监听，处理测试报告和更新状态
     *
     * @param record
     */
    @KafkaListener(topics = {KafkaTopicConfig.REPORT_STATE_TOPIC_NAME}, groupId = "report-test-gp")
    public void onStressReportStateMessage(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        // 打印出消息内容
        System.out.println("消费：" + record.topic() + "-" + record.partition() + "-" + record.value());
        reportService.updateReportState(JsonUtil.json2Obj(record.value().toString(), ReportUpdateReq.class));
        ack.acknowledge();
    }
}
